package com.Parser;


import com.Pojo.Metricbeat;
import com.Pojo.ServerLog;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sink.IISSink;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;


public class MetricbeatJSONKeyValueServerDeserializationSchema implements KafkaDeserializationSchema<String> {

    private static final long serialVersionUID = 1509391548173891946L;
    private static final Logger logger = LoggerFactory.getLogger(MetricbeatJSONKeyValueServerDeserializationSchema.class);



    @Override
    public boolean isEndOfStream(String s) {
        return false;
    }

    @Override
    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        //Flink解析器固定返回String类型 提醒：接口是泛型，具体返回类型可自己定义
        String node = "" ;

        if (record.value() != null) {
            //使用fastjson解析日志处理，获取实际的Filebeat日志然后转为json字符串
            Metricbeat beat = JSON.parseObject(record.value(), Metricbeat.class);
            ServerLog log = new ServerLog();
            //获取时间
            LocalDateTime dateTime = LocalDateTime.parse(beat.getTimestamp(),
                    DateTimeFormatter.ISO_ZONED_DATE_TIME);//2025-02-14T09:26:19.810Z--ISO 8601格式的日期时间字符串
            //传输的时间戳，比中国少8H。需要手动加回8小时再插入数据库
            LocalDateTime new_time = dateTime.plusHours(8);

            log.setVersion(JSONObject.parseObject(beat.getEcs()).getString("version"));//获取beat的版本
            String contextold = new String(record.value(), StandardCharsets.UTF_8);//将字节数组转为字符串
            log.setContext(contextold); //获取原始文本
            log.setDatetime(new_time); //获取度量收集时间
            log.setModule(JSONObject.parseObject(beat.getService()).getString("type"));

            //获取服务器真实IP
            JSONObject jsonhost = JSONObject.parseObject(beat.getHost());
            log.setHost(jsonhost.getString("ip"));
            String[]  iplist = jsonhost.getString("ip").replace("[","").replace("]","").split(",");
            for(String ip : iplist){
//                logger.warn("ip："+ip);
                if(ip.contains(".")){
                    int Maxlen = ip.length();
                    String ip2 = ip.substring(1,Maxlen-1); //去除首尾多余的""
//                    logger.warn("ip："+ip.substring(1,Maxlen-1));
                    if(ip2.split("\\.")[0].equals("192") || ip2.split("\\.")[0].equals("10")){
                        log.setHost(ip2);
                        break;
                    }
                }
            }
            //获取监控项
            JSONObject result = JSONObject.parseObject(beat.getMetricset()); //转为JSON类
            String type = result.getString("name");
            switch (type){
                case "cpu":
                    log.setMetric("system.cpu.total.pct");//cpu使用总的百分比
                    String results = JSONObject.parseObject(beat.getSystem()).getJSONObject("cpu").getJSONObject("total").getString("pct");
                    log.setResult(results);
                    break;

                case "memory":
                    log.setMetric("system.memory.used.pct"); //内存使用百分比
                    String MemoryResult = JSONObject.parseObject(beat.getSystem()).getJSONObject("memory").getJSONObject("used").getString("pct");
                    log.setResult(MemoryResult);
                    break;

                default:
                    log.setResult(null);
                    log.setMetric(null);
//                    throw new Exception("获取不到监控项");

            }

            node = JSONObject.toJSONString(log);
            }

        return node;
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return TypeExtractor.getForClass(String.class); //显式确定返回的数据类型
    }
}
